package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.TableProcess;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * Author: Felix
 * Date: 2021/11/3
 * Desc: 业务数据动态分流实现类
 *      -1.处理广播流数据
 *          >从流中读取配置信息
 *          >将配置信息封装为TableProcess对象
 *          >获取广播状态
 *          >将配置信息放到状态中
 *                 key:sourceTable + ":" + operatorType
 *                 value:TableProcess对象
 *
 *      -2.处理业务流数据
 *          >获取广播状态
 *          >从处理的业务流中获取业务数据库表名以及操作类型  封装为key   sourceTable + ":" + operatorType
 *          >从广播状态中获取当前处理的这条数据  对应的配置信息
 *          >根据配置信息判断是事实还是维度   维度---维度侧输出流      事实---主流
 *
 *      -3.在处理广播流中配置信息的时候，提前创建维度表
 *      -4.字段过滤
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private OutputTag<JSONObject> dimTag;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    private Connection conn;

    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        //获取连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    public TableProcessFunction(OutputTag<JSONObject> dimTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.dimTag = dimTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //处理业务流数据
    @Override
    public void processElement(JSONObject jsonObj, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {

        //获取广播状态
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);

        //获取当前处理的业务数据的表名以及操作类型
        String table = jsonObj.getString("table");
        String type = jsonObj.getString("type");

        //如果对历史数据进行处理   那么使用maxwell的bootstrap同步的时候，类型是bootstrap-insert，我们这里也修改为insert
        if (type.equals("bootstrap-insert")) {
            type = "insert";
            jsonObj.put("type", type);
        }

        //将表名和操作类型 拼接为从状态中获取数据的key
        String key = table + ":" + type;

        //根据key  从广播状态中获取当前处理的这条数据  对应的配置信息
        TableProcess tableProcess = broadcastState.get(key);

        if(tableProcess != null){
            //在配置表中找到了对应的配置

            //不管是事实还是维度     在向下游传递之前  都应该获取目的地
            String sinkTable = tableProcess.getSinkTable();
            jsonObj.put("sink_table",sinkTable);

            //对字段进行过滤
            JSONObject dataJsonObj = jsonObj.getJSONObject("data");
            String sinkColumns = tableProcess.getSinkColumns();
            filterColumn(dataJsonObj,sinkColumns);


            //判断是事实还是维度
            String sinkType = tableProcess.getSinkType();
            if(TableProcess.SINK_TYPE_HBASE.equals(sinkType)){
                //维度数据  --放到维度侧输出流中
                ctx.output(dimTag,jsonObj);
            }else if(TableProcess.SINK_TYPE_KAFKA.equals(sinkType)){
                //事实数据  --放到主流中
                out.collect(jsonObj);
            }
        }else{
            //当前处理的这条业务数据   在配置表中没有找到对应的配置
            System.out.println("no this key in tableProcess :" + key);
        }

    }

    //过滤字段     dataJsonObj :{"tm_name":"ww","logo_url":"qq","id":14}   sinkColumns: "id,tm_name"
    private void filterColumn(JSONObject dataJsonObj, String sinkColumns) {
        //获取保留的字段
        String[] fieldArr = sinkColumns.split(",");
        //为了判断字段的包含关系  将数组转换为集合
        List<String> fieldList = Arrays.asList(fieldArr);

        //遍历json对象中的元素
        Set<Map.Entry<String, Object>> entrySet = dataJsonObj.entrySet();

        entrySet.removeIf(entry->!fieldList.contains(entry.getKey()));

    }

    //处理广播流数据
    @Override
    public void processBroadcastElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {

        //将配置信息的json字符串转换为json对象
        JSONObject jsonObj = JSON.parseObject(jsonStr);
        //将读取到的一条配置信息 封装为TableProcess对象
        TableProcess tableProcess = JSON.parseObject(jsonObj.getString("data"), TableProcess.class);

        //获取业务数据库表名
        String sourceTable = tableProcess.getSourceTable();
        //获取操作类型
        String operateType = tableProcess.getOperateType();
        //获取维度还是事实标记   hbase--维度        kafka---事实
        String sinkType = tableProcess.getSinkType();
        //获取输出的目的地
        String sinkTable = tableProcess.getSinkTable();
        //获取建表主键
        String sinkPk = tableProcess.getSinkPk();
        //获取建表字段  或者叫保留字段
        String sinkColumns = tableProcess.getSinkColumns();
        //获取建表扩展语句
        String sinkExtend = tableProcess.getSinkExtend();

        //判断当前配置数据是不是维度配置
        if(TableProcess.SINK_TYPE_HBASE.equals(sinkType)&&"insert".equals(operateType)){
            //是维度配置  提前创建维度表
            checkTable(sinkTable,sinkColumns,sinkPk,sinkExtend);
        }

        //拼接广播状态中的key
        String key = sourceTable + ":" + operateType;
        //获取广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //将配置信息放到状态中
        broadcastState.put(key,tableProcess);
    }

    //创建维度表
    private void checkTable(String tableName, String fields, String pk, String ext) {
        //空值处理
        if(pk==null){
            pk="id";
        }
        if(ext == null){
            ext = "";
        }

        //拼接建表语句
        StringBuilder createSql = new StringBuilder("create table if not exists "+ GmallConfig.HBASE_SCHEMA +"."+tableName+"(");
        //获取表中所有的字段   id,tm_name
        String[] fieldArr = fields.split(",");
        for (int i = 0; i < fieldArr.length; i++) {
            String fieldName = fieldArr[i];
            if(fieldName.equals(pk)){
                createSql.append(fieldName +" varchar primary key ");
            }else{
                createSql.append(fieldName +" varchar ");
            }
            //判断是不是最后一个字段
            if(i < fieldArr.length -1){
                createSql.append(",");
            }
        }

        createSql.append(")" + ext);
        System.out.println("在phoenix中建表的语句为：" + createSql);

        PreparedStatement ps = null;
        try {
            //创建数据库操作对象
            ps = conn.prepareStatement(createSql.toString());
            //执行sql语句
            ps.execute();
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException("在phoenix中建表失败~~~");
        }finally {
            //释放资源
            if(ps != null){
                try {
                    ps.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}
